package com.fordream.udaf;

import com.aliyun.odps.io.Text;
import com.aliyun.odps.io.Writable;
import com.aliyun.odps.io.WritableUtils;
import com.aliyun.odps.udf.ExecutionContext;
import com.aliyun.odps.udf.UDFException;
import com.aliyun.odps.udf.Aggregator;
import com.aliyun.odps.udf.annotation.Resolve;

import java.util.HashMap;
import java.util.Map;

/**
 * mostFrequentValue 返回聚合结果出现次数最多的值
 */

@Resolve({"string->string"})
public class MostFrequentValueUDAF extends Aggregator {

    private Map<String, Integer> valueCountMap;

    @Override
    public void setup(ExecutionContext ctx) throws UDFException {
        valueCountMap = new HashMap<>();
    }

    @Override
    public Writable newBuffer() {
        return new Text();
    }

    @Override
    public void iterate(Writable buffer, Writable[] args) throws UDFException {
        if (args != null && args.length > 0) {
            if (args[0]==null){
                return;
            }
            String value = args[0].toString(); // Assuming the first argument is a string
            if (value != null&&value!="") {
                valueCountMap.put(value, valueCountMap.getOrDefault(value, 0) + 1);
            }
        }
    }

    @Override
    public void merge(Writable buffer, Writable partial) throws UDFException {
        // Merging logic (not needed for this UDAF)
    }

    @Override
    public Writable terminate(Writable buffer) throws UDFException {
        String mostFrequentValue = null;
        int maxCount = 0;

        for (Map.Entry<String, Integer> entry : valueCountMap.entrySet()) {
            if (entry.getValue() > maxCount) {
                mostFrequentValue = entry.getKey();
                maxCount = entry.getValue();
            }
        }

        return new Text(mostFrequentValue);
    }

    @Override
    public void close() throws UDFException {
        valueCountMap.clear();
    }
}
